/*
 * EventLoop.cc
 *
 *  Created on: 2021年4月5日
 *      Author: san
 */

#include "EventLoop.h"
#include <signal.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <iostream>
#include "base/Logging.h"
#include "Util.h"

class IgnoreSigPipe
{
 public:
  IgnoreSigPipe()
  {
    ::signal(SIGPIPE, SIG_IGN);
    // LOG << "Ignore SIGPIPE";
  }
};

IgnoreSigPipe initObj;

namespace san {

__thread EventLoop *t_loopInThisThread = 0;

int createEventfd()
{
	// 设置不阻塞、fork时关闭
	int eventFd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
	if (eventFd < 0)
	{
		LOG << "get EventFd error";
		abort();
	}
	return eventFd;
}

EventLoop::EventLoop()
	: looping_(false),
	  poller_(new Epoll()),
	  wakeupFd_(createEventfd()),
	  quit_(false),
	  eventHandling_(false),
	  callingPendingFunctors_(false),
	  threadId_(CurrentThread::tid()),
	  pwakeupChannel_(new Channel(this, wakeupFd_))
{
	if (t_loopInThisThread)
	{
//	    LOG << "Another EventLoop " << t_loopInThisThread << " exists in this "
//	    		"thread " << threadId_;
	}
	else
		t_loopInThisThread = this;

	// FIXME: ET or LT
	pwakeupChannel_->setEvents(EPOLLIN | EPOLLET);
	pwakeupChannel_->setReadHandler(std::bind(&EventLoop::handleRead, this));
	pwakeupChannel_->setConnHandler(std::bind(&EventLoop::handleConn, this));
	poller_->epoll_add(pwakeupChannel_, 0);
}

EventLoop::~EventLoop()
{
	close(wakeupFd_);
	t_loopInThisThread = NULL;
}

void EventLoop::wakeup()
{
	// wake 可以从 epoll_wait 中唤醒 loop(), 这样可以使得 dopending_中的函数得到及时的处理
	uint64_t one = 1;
	ssize_t n = writen(wakeupFd_, (char*)(&one), sizeof one);
	if (n != sizeof one)
		LOG << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
}

void EventLoop::handleConn()
{
	// ??? 不太明白
	updatePoller(pwakeupChannel_, 0);
}

void EventLoop::handleRead()
{
	uint64_t one = 1;
	ssize_t n = readn(wakeupFd_, &one, sizeof one);
	if (n != sizeof one)
	{
		LOG << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
	}

	pwakeupChannel_->setEvents(EPOLLIN | EPOLLET);
}


void EventLoop::runInLoop(Functor &&cb)
{
	if (isInLoopThread())
		cb();
	else
		queueInLoop(std::move(cb));
}

void EventLoop::queueInLoop(Functor &&cb)
{
	{
		MutexLockGuard lock(mutex_);
		pendingFunctors_.emplace_back(std::move(cb));
	}

	if (!isInLoopThread() || callingPendingFunctors_) wakeup();
}

void EventLoop::loop()
{
	assert(!looping_);
	assert(isInLoopThread());
	looping_ = true;
	quit_ = false;
	std::vector<ChannelPtr> ret;
	while (!quit_)
	{
		ret.clear();
		ret = poller_->poll();
		eventHandling_ = true;
		for (auto& it : ret) it->handleEvents();
		eventHandling_ = false;
		doPendingFunctors();
		poller_->handleExpired();
	}
	looping_ = false;
}

void EventLoop::doPendingFunctors()
{
	std::vector<Functor> functors;
	callingPendingFunctors_ = true;

	// CopyOnWrite
	{
		MutexLockGuard lock(mutex_);
		functors.swap(pendingFunctors_);
	}

	for (size_t i = 0; i < functors.size(); ++i)
		functors[i]();
	callingPendingFunctors_ = false;
}

void EventLoop::quit()
{
	quit_ = true;
	if (!isInLoopThread())
	{
		wakeup();
	}
}



} /* namespace san */
